package com.diy.sigmund.diyorder.test.controller;

import com.diy.sigmund.diycommon.entity.ResultMsg;
import com.diy.sigmund.diycommon.spring.config.PortConfig;
import com.diy.sigmund.diycommon.test.entity.MqTestVO;
import com.diy.sigmund.diycommon.util.json.JacksonUtil;
import com.diy.sigmund.diycommon.util.seq.SeqUtil;
import com.diy.sigmund.diyorder.common.middleware.mq.RocketMQProtocolEnum;
import com.diy.sigmund.diyorder.common.middleware.mq.SecKillChargeOrderProducer;
import com.diy.sigmund.diyorder.common.middleware.mq.rocketmq.spring.one.ExtRocketMQTemplate;
import com.diy.sigmund.diyorder.common.middleware.mq.rocketmq.spring.one.ProductWithPayload;
import com.diy.sigmund.diyorder.entity.po.OdSeckillOrder;
import com.fasterxml.jackson.core.type.TypeReference;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.LongAdder;
import javax.annotation.Resource;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQLocalRequestCallback;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer.DefaultMessageListenerConcurrently;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author ylm-sigmund
 * @since 2021/4/17 15:56
 */
@RestController
@RequestMapping("/mqTest")
public class MqTestController {

    private static final Logger log = LoggerFactory.getLogger(MqTestController.class);

    @Autowired
    private SecKillChargeOrderProducer producer;
    @Autowired
    private ExtRocketMQTemplate extRocketMQTemplate;
    @Resource
    private RocketMQTemplate rocketMQTemplate;
    @Value("${demo.rocketmq.genericRequestTopic}")
    private String genericRequestTopic;
    @Autowired
    private PortConfig portConfig;
    private static final LongAdder LONG_ADDER = new LongAdder();

    @PostMapping("/send")
    public ResultMsg send() throws Exception {
        final DefaultMQProducer producer = this.producer.getProducer();
        MqTestVO mqTestVo = new MqTestVO();
        mqTestVo.setUserId(UUID.randomUUID().toString());
        mqTestVo.setPhoneNum(UUID.randomUUID().toString().substring(0, 11));
        mqTestVo.setCreateTime(new Date());
        final Message message = new Message(RocketMQProtocolEnum.SECKILL_TEST_TOPIC.getTopic(),
                JacksonUtil.toJson(mqTestVo).getBytes(StandardCharsets.UTF_8));
        SendResult sendResult = producer.send(message);
        log.info("sendResult:{}", sendResult);
        return ResultMsg.success(mqTestVo);
    }

    @PostMapping("/sendOrderly")
    public ResultMsg sendOrderly() throws Exception {
        LONG_ADDER.increment();

        MqTestVO mqTestVo = new MqTestVO();
        mqTestVo.setUserId(UUID.randomUUID().toString());
        mqTestVo.setName("ylm-sigmund+" + portConfig.getPort() + "+" + LONG_ADDER.intValue());
        mqTestVo.setPhoneNum(UUID.randomUUID().toString().substring(0, 11));
        mqTestVo.setCreateTime(new Date());
        final Message message = new Message(RocketMQProtocolEnum.SECKILL_TEST_TOPIC.getTopic(),
                JacksonUtil.toJson(mqTestVo).getBytes(StandardCharsets.UTF_8));

        final DefaultMQProducer producer = this.producer.getProducer();
        for (int i = 0; i < 1; i++) {
            SendResult sendResult = producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    log.info("msg:{}, arg:{}", new String(msg.getBody()), arg);
                    int value = arg.hashCode() % mqs.size();
                    if (value < 0) {
                        value = Math.abs(value);
                    }
                    log.info("INDEX:{}", value);
                    return mqs.get(value);
                }
            }, "sendOrderly test!",600000);
            log.info("sendResult:{}", sendResult);
        }
        return ResultMsg.success(mqTestVo);
    }

    /**
     * 有异常会返回ConsumeConcurrentlyStatus.RECONSUME_LATER，否则返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS
     *
     * @see DefaultMessageListenerConcurrently#consumeMessage(java.util.List, org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext)
     */
    @PostMapping("/sendBoot")
    public ResultMsg sendBoot() {
        MqTestVO mqTestVo = new MqTestVO();
        mqTestVo.setUserId("111");
        mqTestVo.setPhoneNum(UUID.randomUUID().toString().substring(0, 11));
        // mqTestVo.setCreateTime(new Date());
        // 普通消息
        // rocketMQTemplate.convertAndSend("sendDataType:*", mqTestVo, null, null);
        // 同步消息
        SendResult sendMessage = rocketMQTemplate.syncSend("sendDataType:*", mqTestVo);
        log.info("syncSend sendMessage:{}", sendMessage);

        // GenericMessage<MqTestVO> mqTestVOGenericMessage = new GenericMessage<>(mqTestVo);
        // SendResult sendResult = rocketMQTemplate.syncSend("sendDataType:*", mqTestVOGenericMessage);
        // log.info("mqTestVOGenericMessage:{}", mqTestVOGenericMessage);
        // log.info("syncSend mqTestVOGenericMessage:{}", sendResult);

        // 异步消息
        // rocketMQTemplate.asyncSend("sendDataType:*", mqTestVo, new SendCallback() {
        //     @Override
        //     public void onSuccess(SendResult sendResult) {
        //         log.info("asyncSend success {}", sendResult);
        //     }
        //
        //     @Override
        //     public void onException(Throwable e) {
        //         log.error("asyncSend onException", e);
        //     }
        // });

        // 单向消息
        // rocketMQTemplate.sendOneWay("sendDataType:*", mqTestVo);

        return ResultMsg.success(mqTestVo);
    }

    /**
     * <br>发送延迟消息
     * <br>基本概念：延迟消息是指生产者发送消息发送消息后，不能立刻被消费者消费，需要等待指定的时间后才可以被消费。
     * <br>场景案例：用户下了一个订单之后，需要在指定时间内(例如30分钟)进行支付，在到期时用户仍未支付可以取消订单。
     * <br>Broker在启动时，内部会创建一个内部主题：SCHEDULE_TOPIC_XXXX，根据延迟level的个数，创建对应数量的队列，也就是说18个level对应了18个队列。
     * <br>注意，这并不是说这个内部主题只会有18个队列，因为Broker通常是集群模式部署的，因此每个节点都有18个队列。
     * <br>延迟级别的值可以进行修改，以满足自己的业务需求，可以修改/添加新的level。例如：你想支持2天的延迟，修改最后一个level的值为2d，这个时候依然是18个level；
     * <br>也可以增加一个2d，这个时候总共就有19个level。
     * <br>messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     * <br>
     * <br>延迟消息在RocketMQ Broker端的流转:
     * <br>修改消息Topic名称和队列信息
     * <br>转发消息到延迟主题的CosumeQueue中
     * <br>延迟服务消费SCHEDULE_TOPIC_XXXX消息
     * <br>将信息重新存储到CommitLog中
     * <br>将消息投递到目标Topic中
     * <br>消费者消费目标topic中的数据
     * <br>
     * <br>延迟消息与消费重试的关系:
     * <br>消息重试的16个级别，实际上是把延迟消息18个级别的前两个level去掉了。
     * <br>事实上，RocketMQ的消息重试也是基于延迟消息来完成的。在消息消费失败的情况下，将其重新当做延迟消息投递回Broker。
     * <br>在投递回去时，会跳过前两个level，因此只重试16次。
     *
     * @see <a href="https://cloud.tencent.com/developer/article/1581368">深入理解RocketMQ延迟消息</a>
     */
    @PostMapping("/sendDelay")
    public ResultMsg sendDelay() {
        MqTestVO mqTestVo = new MqTestVO();
        mqTestVo.setUserId("111");
        mqTestVo.setPhoneNum(UUID.randomUUID().toString().substring(0, 11));

        // 使用 timeout 和 delayLevel 参数参数以同步模式发送请求并接收通用类型的回复。
        ProductWithPayload<String> replyGenericObject = rocketMQTemplate.sendAndReceive(genericRequestTopic,
                "request generic",
                new TypeReference<ProductWithPayload<String>>() {
                }.getType(),
                30000,
                0);
        // System.out.printf("send %s and receive %s %n", "request generic", replyGenericObject);
        // rocketMQTemplate.sendAndReceive(genericRequestTopic,
        //         "request generic",
        //         new RocketMQLocalRequestCallback() {
        //             @Override
        //             public void onSuccess(Object message) {
        //                 log.info("------- ConsumerWithReplyGeneric received: {}", message);
        //             }
        //
        //             @Override
        //             public void onException(Throwable e) {
        //                 log.error("------- ConsumerWithReplyGeneric onException", e);
        //             }
        //         },
        //         30000,
        //         3
        // );
        return ResultMsg.success(mqTestVo);
    }


    /**
     * <br>RocketMQ在其消息定义的基础上，对事务消息扩展了两个相关的概念：
     * <br>Half(Prepare) Message——半消息(预处理消息)
     * <br>半消息是一种特殊的消息类型，该状态的消息暂时不能被Consumer消费。当一条事务消息被成功投递到Broker上，但是Broker并没有接收到Producer发出的二次确认时，该事务消息就处于"暂时不可被消费"状态，该状态的事务消息被称为半消息。
     * <br>Message Status Check——消息状态回查
     * <br>由于网络抖动、Producer重启等原因，可能导致Producer向Broker发送的二次确认消息没有成功送达。如果Broker检测到某条事务消息长时间处于半消息状态，则会主动向Producer端发起回查操作，查询该事务消息在Producer端的事务状态(Commit 或 Rollback)。可以看出，Message Status Check主要用来解决分布式事务中的超时问题。
     *
     * @see <a href="https://blog.csdn.net/weixin_34452850/article/details/88851419">详解RocketMQ事务消息</a>
     */
    @PostMapping("/sendMessageInTransaction/{arg}")
    public ResultMsg sendMessageInTransaction(@PathVariable String arg) throws Exception {
        // 事务消息
        OdSeckillOrder odSeckillOrder = new OdSeckillOrder();
        org.springframework.messaging.Message<OdSeckillOrder> order = MessageBuilder.withPayload(odSeckillOrder)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, SeqUtil.snowflakeId())
                .build();
        // RMQ_SYS_TRANS_HALF_TOPIC 这个topic可以查询到消息
        // 分别运行TransactionListenerImpl和ProductListener，可以看出只有事务执行成功的订单才会通知商品服务进行减库存。
        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction("transactionMessage:*",
                order, arg);
        // transactionSendResult.getLocalTransactionState();
        // transactionSendResult.setLocalTransactionState(LocalTransactionState.COMMIT_MESSAGE);
        // transactionSendResult.setLocalTransactionState(LocalTransactionState.ROLLBACK_MESSAGE);
        log.info("transactionSendResult:{}", JacksonUtil.toJson(transactionSendResult));
        return ResultMsg.success();
    }
}
